package org.shj.javaconcurrencypattern.pipeline;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;

public class Main {

	public static void main(String[] args) throws Exception{
		
		SimplePipeline<String, String> pipeline = new SimplePipeline<String, String>();
		
		BlockingQueue<List<String>> queue = new LinkedBlockingQueue<List<String>>();
		ExecutorService excutorService = Executors.newFixedThreadPool(2);
		
		Pipe<String, List<String>> order = new AbstractPipe<String, List<String>>(){
			protected List<String> doProcess(String input) throws PipeException {
				System.out.println(input + " order: ");
				List<String> list = new ArrayList<String>();
				for(int i = 0; i < 5 ; i++){
					list.add(input + "'s cai" + i);
					System.out.println("cai" + i);
				}
				return list;
			}
			
		};
		
		pipeline.addPipe(order);
		
		Pipe<List<String>, List<String>> chef = new AbstractParallelPipe<List<String>, List<String>, String>(queue, excutorService){
			protected List<Callable<String>> buildTasks(List<String> input) throws Exception {
				List<Callable<String>> list = new ArrayList<Callable<String>>();
				for(final String cai : input){
					list.add(new Callable<String>(){
						public String call() throws Exception {
							System.out.println(Thread.currentThread().getName() + " is cooking " + cai);
							return "Finished " + cai;
						}
					});
				}
				return list;
			}

			protected List<String> combineResults(List<Future<String>> subTaskResults) throws Exception {
				List<String> list = new ArrayList<String>();
				for(Future<String> future : subTaskResults){
					list.add(future.get());
				}
				return list;
			}
			
		};
		
		pipeline.addPipe(chef);
		
		Pipe<List<String>, Void> servant = new AbstractPipe<List<String>, Void>(){
			protected Void doProcess(List<String> input) throws PipeException {
				for(String s : input){
					System.out.println("Give customer: " + s);
				}
				return null;
			}
			
		};
		
		pipeline.addPipe(servant);
		
		pipeline.init(pipeline.newDefaultPipelineContext());
		
		pipeline.process("customer1");
		
	}
}
